package cn.lagou.graphx

import breeze.signal.OptWindowFunction.User
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Airports {

  case class User(name: String, inDeg: Int, outDeg: Int)


  def main(args: Array[String]): Unit = {
    // 初始化
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")

    // 初始化数据
    // 定义顶点 (Long, info)
    val vertexArray: Array[(VertexId, String)] = Array(
      (1L, "SFO"),
      (2L, "ORD"),
      (3L, "DFW")
    )

    // 定义边 (Long, Long, attr)
    val edgeArray: Array[Edge[Int]] = Array(
      Edge(1L, 2L, 1800),
      Edge(2L, 3L, 800),
      Edge(3L, 1L, 1400)
    )

    //飞机航线按直线说应该是双方向的一条直线，暂不考虑给边加方向
    /*
    val edgeArray2 = edgeArray.flatMap{
      edge =>
      val goback = Array(Edge(edge.srcId, edge.dstId,edge.attr), Edge(edge.dstId, edge.srcId,edge.attr))
        goback
    }
    */

    // 构造vertexRDD和edgeRDD
    val vertexRDD: RDD[(Long, String)] = sc.makeRDD(vertexArray)
    val edgeRDD: RDD[Edge[Int]] = sc.makeRDD(edgeArray)

    // 构造图Graph[VD,ED]
    val graph: Graph[String, Int] = Graph(vertexRDD, edgeRDD)

    //    graph.pageRank()
    println("*********** 所有的顶点 ***********")
    // 找出图中年龄大于30的顶点
    graph.vertices
      .foreach(println)
    println(s"顶点数 = ${graph.vertices.count()}")
    println("*********** 所有的边 ***********")
    graph.edges
      .filter{edge => edge.attr>5}
      .foreach(println)
    println(s"边数 = ${graph.edges.count()}")
    println("*********** 距离大于1000的 ***********")
    graph.edges
      .filter{edge => edge.attr>1000}
      .sortBy{_.attr}
      .foreach(println)
    println("*********** 所有的tripltes ***********")
    // 列出边属性>5的tripltes
    graph.triplets
      //.filter(t => t.attr > 5)
      .foreach(println)

    /*
    // 转换操作
    // 顶点的转换操作。所有人的年龄加 10 岁
    graph.mapVertices{case (id, (name, age)) => (id, (name, age+10))}
      .vertices
      .foreach(println)

    // 边的转换操作。边的属性*2
    graph.mapEdges(e => e.attr*2)
      .edges
      .foreach(println)

    // 结构操作
    // 顶点年龄 > 30 的子图
    val subGraph: Graph[(String, Int), Int] = graph.subgraph(vpred = (id, vd) => vd._2 >= 30)
    println("************** 子图 ***************")
    subGraph.edges.foreach(println)
    subGraph.vertices.foreach(println)

    // 连接操作
    println("*********************** 连接操作 ***********************")
    // 创建一个新图，顶点VD的数据类型为User，并从graph做类型转换
    val initialUserGraph: Graph[User, Int] = graph.mapVertices { case (_, (name, age)) => User(name, 0, 0) }

    // initialUserGraph与inDegrees、outDegrees 进行 join，修改 inDeg、outDeg
    val userGraph: Graph[User, Int] = initialUserGraph.outerJoinVertices(initialUserGraph.inDegrees) {
      case (id, u, inDegOpt) => User(u.name, inDegOpt.getOrElse(0), u.outDeg)
    }.outerJoinVertices(initialUserGraph.outDegrees) {
      case (id, u, outDegOpt) => User(u.name, u.inDeg, outDegOpt.getOrElse(0))
    }

    userGraph.vertices.foreach(println)

    // 找到 出度=入度 的人员
    userGraph.vertices.filter { case (id, u) => u.inDeg == u.outDeg }
      .foreach(println)

    // 聚合操作
    // 找出5到各顶点的最短距离
    val sourceId: VertexId = 5L // 定义源点
    val initialGraph: Graph[Double, Int] = graph.mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity)

    val sssp: Graph[Double, Int] = initialGraph.pregel(Double.PositiveInfinity)(
      // 两个消息来的时候，取它们当中路径的最小值
      (id, dist, newDist) => math.min(dist, newDist),

      // Send Message函数
      // 比较 triplet.srcAttr + triplet.attr和 triplet.dstAttr。如果小于，则发送消息到目的顶点
      triplet => { // 计算权重
        if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
          Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
        } else {
          Iterator.empty
        }
      },

      // mergeMsg
      (a, b) => math.min(a, b) // 最短距离
    )

    println("找出5到各顶点的最短距离")
    println(sssp.vertices.collect.mkString("\n"))

     */
  }
}
